package processer

import (
	"fmt"
	"hermes/manager"
	"hermes/message"
)

func TableFilterMessage(message *message.ProcessMessage) bool {
	var skip = TableFilterMessageByName(message.TableName)
	return skip
}

func TableFilterMessageByName(tableName string) bool  {
	//schemaList := []string{ "b2b5.tb_hermes_raw","b2b5.tb_hermes_test"}
	//var skip = true
	//for _, v:= range schemaList {
	//	if strings.Contains(tableName,v){
	//		skip = false
	//	}
	//}
	var skip = manager.IsTableFilter(tableName)
	return skip
}

func ColumnsFilterMessage(message *message.ProcessMessage) bool {
	if message.Operation == "UPDATE" {
		tableFilterMessage := manager.GetTableFilterMessage(message.TableName)
		if len(tableFilterMessage.CareColumns) > 0 {
			var columnIndex []int
			for _,careColumnName := range tableFilterMessage.CareColumns {
				fmt.Println(careColumnName)
				for i,columnName :=range message.TableColumnNames {
					if careColumnName == columnName {
						columnIndex = append(columnIndex,i)
					}
				}
			}
			handleColumnChange(message,columnIndex)
			if len(message.Rows) == 0{
				return true
			}
		}
	}
	return false
}

func handleColumnChange(message *message.ProcessMessage,columnIndex[] int) {

	var beforeRow []interface{}
	var afterRow []interface{}
	var resultRows  [][]interface{}
	for rowIndex,row :=range message.Rows{
		if rowIndex % 2 ==0 {
			beforeRow = row
		}else{
			afterRow = row
			if isRowDiff(beforeRow,afterRow,columnIndex){
				resultRows = append(resultRows,beforeRow,afterRow)
			}
		}
	}
	message.Rows = resultRows
}

func isRowDiff(beforeRow []interface{},afterRow []interface{},columnIndexs []int) bool {
	for _,columnIndex:=range columnIndexs {
		if beforeRow[columnIndex] != afterRow[columnIndex] {
			return true
		}
	}
	return false
}